package com.boarsoft.message.rocketmq;

import java.io.Closeable;
import java.util.List;

import org.apache.rocketmq.client.consumer.MQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RocketMQConsumer implements MessageListenerConcurrently, Closeable {
	private static final Logger log = LoggerFactory.getLogger(RocketMQConsumer.class);

	/** */
	protected MQPushConsumer consumer;
	/** */
	protected MessageListenerConcurrently listener;
	/** */
	protected String tag;

	public void init(String code) {
		consumer.registerMessageListener(this); // 注册消费者监听
		try {
			// 第二个参数表示消费匹配的tag * 表示topic所有的tag
			consumer.subscribe(code, tag);
			consumer.start();
			log.info("RocketMQ consumer {} started", consumer);
		} catch (MQClientException e) {
			log.error("Error on start RocketMQ consumer {}", consumer, e);
		}
	}

	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
		if (listener != null) {
			listener.consumeMessage(msgs, context);
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
		return ConsumeConcurrentlyStatus.RECONSUME_LATER;
	}

	@Override
	public void close() {
		consumer.shutdown();
		consumer = null;
	}

	public MQPushConsumer getConsumer() {
		return consumer;
	}

	public MessageListener getListener() {
		return listener;
	}

	public void setConsumer(MQPushConsumer consumer) {
		this.consumer = consumer;
	}

	public void setListener(MessageListenerConcurrently listener) {
		this.listener = listener;
	}

	public String getTag() {
		return tag;
	}

	public void setTag(String tag) {
		this.tag = tag;
	}
}
